package org.easy.tool.common;

import com.google.common.collect.Lists;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * @author Valley
 * Created by Valley on 2023/11/1 14:32
 */
public class ThreadUtil {

    public static final Object LOCK = new Object();
    public static LinkedHashMap<String, Object> lockMap = new LinkedHashMap<>();

    /**
     * 功能：从lockMap中取出指定obj锁
     */
    public static Object getLockFromMap(String lockId) {
        return lockMap.computeIfAbsent(lockId, k -> new Object());
    }

    /**
     * 功能：使运行的指定线程等待
     */
    public static void lock(String lockId) throws Exception{
        Object lockFromMap = getLockFromMap(lockId);
        synchronized (lockFromMap){
            lockFromMap.wait();
        }
    }

    /**
     * 功能：使运行的指定线程苏醒
     */
    public static void unlock(String lockId) throws Exception{
        Object lockFromMap = getLockFromMap(lockId);
        synchronized (lockFromMap){
            lockFromMap.notify();
        }
    }

    /**
     * 功能：使当前运行的线程等待
     */
    public static void lock() throws Exception{
        synchronized (LOCK){
            LOCK.wait();
        }
    }

    /**
     * 功能：使当前沉睡等待的线程苏醒
     */
    public static void unlock() throws Exception{
        synchronized (LOCK){
            LOCK.notify();
        }
    }

    /**
     * 功能：封装执行批量异步程序
     */
    public static List<Object> scheduleTask(List taskList, Integer maxTask, Function function) throws Exception{
        //最大线程等于物理机最高性能
        int thread = Runtime.getRuntime().availableProcessors();
        if (maxTask == null || maxTask > thread) maxTask = thread;
        ExecutorService service = Executors.newFixedThreadPool(maxTask);
        //控制任务数变化
        ConcurrentLinkedQueue<Future> futureList = new ConcurrentLinkedQueue<>();
        ConcurrentLinkedQueue<Object> resultList = new ConcurrentLinkedQueue<>();
        //添加任务
        for (Object task : taskList) {
            Future<?> future = service.submit(()->function.apply(task));
            futureList.add(future);
        }
        //轮询直到任务全部执行完毕
        while (true){
            for (Future future : futureList) {
                //isDone()包括 完成 取消 异常中止
                if (future.isDone()){
                    Long time = (Long) future.get();
                    resultList.add(time);
                    futureList.remove(future);
                }
            }
            //任务执行完毕，退出
            if (futureList.size() == 0) break;
            // 每次轮询休息1秒，避免CPU占用
            Thread.sleep(1000);
        }
        return new ArrayList<>(resultList);
    }



}
